package com.example.flinkcourse.lesson4.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.functions.source.Source;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;

/**
 * 文件数据源工厂类
 * 用于创建文件数据源，支持持续监控文件变化
 */
public class FileSourceFactory {
    private static final Logger LOG = LoggerFactory.getLogger(FileSourceFactory.class);

    /**
     * 创建文件数据源
     * @param inputPath 输入文件路径
     * @param watchInterval 监控间隔（毫秒）
     * @return 文件数据源
     */
    public static Source<String> createFileSource(String inputPath, long watchInterval) {
        LOG.info("Creating file source for path: {}", inputPath);
        
        FileSource<String> source = FileSource
            .forRecordStreamFormat(
                new TextLineInputFormat(),
                new Path(inputPath)
            )
            .monitorContinuously(Duration.ofMillis(watchInterval))
            .build();
        
        LOG.info("File source created successfully");
        return source;
    }
    
    /**
     * 创建单次读取的文件数据源
     * @param inputPath 输入文件路径
     * @return 文件数据源
     */
    public static Source<String> createOneTimeFileSource(String inputPath) {
        LOG.info("Creating one-time file source for path: {}", inputPath);
        
        FileSource<String> source = FileSource
            .forRecordStreamFormat(
                new TextLineInputFormat(),
                new Path(inputPath)
            )
            .build();
        
        LOG.info("One-time file source created successfully");
        return source;
    }
} 